package kim.artemis.rpc.core.protocol.client;

import io.netty.bootstrap.Bootstrap;
import io.netty.channel.Channel;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.EventLoopGroup;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import kim.artemis.rpc.core.common.ArtemisDecoder;
import kim.artemis.rpc.core.common.ArtemisEncoder;
import kim.artemis.rpc.core.model.ArtemisRpcRequest;
import kim.artemis.rpc.core.model.ArtemisRpcResponse;
import kim.artemis.rpc.core.protocol.serializer.ArtemisSerializer;
import lombok.extern.slf4j.Slf4j;


/**
 * 客户端会话
 * @author Wq.
 */
@Slf4j
public class ArtemisRpcClientSession {

    private EventLoopGroup loopGroup;
    private Channel channel;

    public void initClient (String socketAddress, ArtemisSerializer serializer, final ArtemisRpcClientInvoker invokerFactory) throws Exception {
        String[] socketAddressArr = socketAddress.trim().split(":");
        String host = socketAddressArr[0];
        int port = Integer.valueOf(socketAddressArr[1]);
        loopGroup = new NioEventLoopGroup();
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(loopGroup)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        socketChannel.pipeline()
                                // 序列化请求
                                .addLast(new ArtemisEncoder(ArtemisRpcRequest.class, serializer))
                                // 返序列化响应结果
                                .addLast(new ArtemisDecoder(ArtemisRpcResponse.class, serializer))
                                // 处理程序
                                .addLast(new ArtemisRpcClientHandler(invokerFactory));
                    }
                })
                .option(ChannelOption.TCP_NODELAY, true)
                .option(ChannelOption.SO_KEEPALIVE, true)
                .option(ChannelOption.CONNECT_TIMEOUT_MILLIS, 10000);
        this.channel = bootstrap.connect(host, port).sync().channel();

        // 验证
        if (!isActive()) {
            close();
            return;
        }
    }

    public boolean isActive() {
        if (this.channel != null) {
            return this.channel.isActive();
        }
        return false;
    }

    public void close() {
        if (this.channel != null && this.channel.isActive()) {
            this.channel.close();        // if this.channel.isOpen()
        }
        if (this.loopGroup != null && !this.loopGroup.isShutdown()) {
            this.loopGroup.shutdownGracefully();
        }
        log.debug(">>>>>>> artemis rpc, client close.");
    }

    public void send(ArtemisRpcRequest artemisRpcRequest) throws Exception {
        this.channel.writeAndFlush(artemisRpcRequest).sync();
    }

}
